package portrait

import java.sql.{DriverManager, ResultSet}

import Configer.Configer
import ch.hsr.geohash.GeoHash
import org.apache.commons.lang.StringUtils
import org.apache.spark.sql.Row
import scalikejdbc.{DB, SQL}
import utils.TurnType

object BusinessTags extends Tags {
  override def makeTags(any: Any*): List[(String, Int)] = {
    val row = any(0).asInstanceOf[Row]
    var list = List[(String, Int)]()
    val lat = TurnType.toDouble(row.getAs[String]("lat"))
    val long = TurnType.toDouble(row.getAs[String]("long"))
    val geohash = GeoHash.withCharacterPrecision(lat, long, 8).toBase32
    //    val conn = DriverManager.getConnection(Configer.url,Configer.user,Configer.password)
    //    val statement = conn.prepareStatement("select business from area34 where geohash=?")
    //    statement.setString(1,geohash)
    //    val set: ResultSet = statement.executeQuery()
    //    if (set.next()){
    //      val bus = set.getString("business")
    //      if (StringUtils.isNotEmpty(bus)) list :+= ("BS"+bus,1)
    //    }
    val tuples: List[String] = DB.readOnly { implicit session =>
      SQL("select business from area34 where geohash=?").bind(geohash)
        .map(tp =>
          tp.string("business")
        ).list().apply()
    }
    tuples.filter(tp=>StringUtils.isNotEmpty(tp)).foreach(tp=>list :+= ("BS"+tp,1))
    list
  }
}
